import hashlib

import mysql.connector
import psycopg2
from psycopg2.extras import DictCursor

# mysql_host="172.16.1.27",
# mysql_port=3306,
# mysql_user="root",
# mysql_password="hmgf@600987",
# mysql_database="hm_erp",
# pg_host="172.16.1.21",
# pg_port=31003,
# pg_user="postgres",
# pg_password="postgres",
# pg_database="upms"

# 数据库配置
MYSQL_CONFIG = {
    "host": "172.16.1.27",
    "port": 3306,
    "user": "root",
    "password": "hmgf@600987",
    "database": "hm_erp"
}

PG_CONFIG = {
    "host": "172.16.1.21",
    "port": 31003,
    "user": "postgres",
    "password": "postgres",
    "database": "upms"
}

import mysql.connector
import psycopg2
import hashlib
import os
from datetime import datetime


# 类型映射表
TYPE_MAPPING = {
    'int': 'integer',
    'tinyint': 'smallint',
    'decimal': 'numeric',
    'varchar': 'text',
    'datetime': 'timestamp'
}


# 数据库连接函数
def connect_mysql():
    try:
        return mysql.connector.connect(**MYSQL_CONFIG)
    except mysql.connector.Error as err:
        print(f"MySQL连接失败: {err}")
        exit(1)


def connect_pg():
    try:
        return psycopg2.connect(**PG_CONFIG)
    except psycopg2.OperationalError as err:
        print(f"PostgreSQL连接失败: {err}")
        exit(1)


# 表结构对比相关函数
def get_mysql_tables(conn):
    """获取MySQL表列表并统一小写"""
    try:
        cursor = conn.cursor()
        cursor.execute("SHOW FULL TABLES WHERE Table_type = 'BASE TABLE'")
        # for row in cursor.fetchall():
        #     print("row:" + row[0])
        return [row[0].lower() for row in cursor.fetchall()]
    except mysql.connector.Error as err:
        print(f"获取MySQL表失败: {err}")
        return []


def get_pg_tables(conn):
    """获取PostgreSQL表列表并统一小写"""
    try:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT LOWER(table_name)
                FROM information_schema.tables 
                WHERE table_schema NOT IN ('pg_catalog', 'information_schema')
                  AND table_type = 'BASE TABLE'
            """)
            # for row in cur.fetchall():
            #     print(row[0])
            return [row[0] for row in cur.fetchall()]
    except Exception as e:
        print(f"获取PostgreSQL表失败: {e}")
        return []


def format_columns(columns):
    """将列信息格式化为可读字符串"""
    return "\n".join([
        f"  - {name}: {info['type']}{' NOT NULL' if not info['nullable'] else ''} "
        f"(默认值: {info['default'] or '无'})"
        for name, info in columns.items()
    ])


def compare_columns(mysql_cols, pg_cols):
    """深度对比列定义"""
    for col in mysql_cols:
        if col not in pg_cols:
            return False
        if (mysql_cols[col]['type'] != pg_cols[col]['type'] or
                mysql_cols[col]['nullable'] != pg_cols[col]['nullable']):
            return False
    return True


def get_mysql_columns(conn, table):
    try:
        cursor = conn.cursor(dictionary=True)
        cursor.execute(f"SHOW FULL COLUMNS FROM `{table}`")
        columns = {}
        for col in cursor:
            base_type = col['Type'].lower()
            pg_type = TYPE_MAPPING.get(base_type, base_type)
            columns[col['Field'].lower()] = {
                'type': pg_type,
                'nullable': col['Null'] == 'YES',
                'default': col['Default']
            }
        return columns
    except Exception as e:
        print(f"获取MySQL表 {table} 列信息失败: {e}")
        return {}


def get_pg_columns(conn, table):
    try:
        with conn.cursor() as cur:
            cur.execute("""
                SELECT column_name, udt_name, is_nullable, column_default
                FROM information_schema.columns
                WHERE LOWER(table_name) = %s
            """, (table.lower(),))
            return {
                row[0].lower(): {
                    'type': row,
                    'nullable': row == 'YES',
                    'default': row
                } for row in cur.fetchall()
            }
    except Exception as e:
        print(f"获取PostgreSQL表 {table} 列信息失败: {e}")
        return {}


def compare_schemas(mysql_conn, pg_conn):
    diff_report = []
    mysql_tables = get_mysql_tables(mysql_conn)
    print("mysql_tables", mysql_tables)
    pg_tables = get_pg_tables(pg_conn)
    print("pg_tables", pg_tables)

    # 检查缺失表
    missing_tables = [t for t in mysql_tables if t not in pg_tables]
    if missing_tables:
        diff_report.append("目标库缺失表: " + ", ".join(missing_tables))

    # 对比共有表结构
    common_tables = set(mysql_tables) & set(pg_tables)
    for table in common_tables:
        mysql_cols = get_mysql_columns(mysql_conn, table)
        pg_cols = get_pg_columns(pg_conn, table)

        if not compare_columns(mysql_cols, pg_cols):
            diff_report.append(f"表 {table} 列定义不一致")
            diff_report.append("MySQL列定义:\n" + format_columns(mysql_cols))
            diff_report.append("PostgreSQL列定义:\n" + format_columns(pg_cols))

    return diff_report


# 数据对比相关函数
def get_mysql_primary_keys(conn, table):
    try:
        cursor = conn.cursor()
        cursor.execute(f"SHOW KEYS FROM `{table}` WHERE Key_name = 'PRIMARY'")
        return [row[4] for row in cursor.fetchall()]  # 列名在第4个位置
    except Exception as e:
        print(f"获取MySQL表 {table} 主键失败: {e}")
        return []


def normalize_data(row):
    """统一数据类型处理"""
    return tuple(
        str(val).lower() if isinstance(val, bool) else val
        for val in row.values()
    )


def fetch_mysql_data(cursor, table, pk_columns, limit, offset):
    try:
        print(f"""
            SELECT * FROM `{table}`
            ORDER BY {','.join(pk_columns)}
            LIMIT {limit} OFFSET {offset}
        """)
        cursor.execute(f"""
            SELECT * FROM `{table}`
            ORDER BY {','.join(pk_columns)}
            LIMIT {limit} OFFSET {offset}
        """)
        return {tuple(row[col] for col in pk_columns): normalize_data(row) for row in cursor}
    except Exception as e:
        print(f"查询MySQL表 {table} 失败: {e}")
        return {}


def fetch_pg_data(cursor, table, pk_columns, limit, offset):
    try:
        cursor.execute(f"""
            SELECT * FROM "{table}"
            ORDER BY {','.join([f'"{col}"' for col in pk_columns])}
            LIMIT {limit} OFFSET {offset}
        """)
        return {tuple(row[col] for col in pk_columns): normalize_data(row) for row in cursor}
    except psycopg2.errors.UndefinedTable:
        print(f"错误：PostgreSQL表 {table} 不存在")
        return None
    except Exception as e:
        print(f"查询PostgreSQL表 {table} 失败: {e}")
        return None


def hash_data(data):
    return hashlib.md5(str(data).encode()).hexdigest()


def find_data_diff(table, mysql_batch, pg_batch, pk_columns):
    diff = []
    for pk in set(mysql_batch.keys()) | set(pg_batch.keys()):
        mysql_row = mysql_batch.get(pk)
        pg_row = pg_batch.get(pk)

        if mysql_row and not pg_row:
            diff.append(f"行丢失 (PG缺失) PK={pk}")
        elif pg_row and not mysql_row:
            diff.append(f"行多余 (PG存在) PK={pk}")
        elif mysql_row != pg_row:
            for idx, (m_val, p_val) in enumerate(zip(mysql_row, pg_row)):
                if m_val != p_val:
                    col = pk_columns[idx % len(pk_columns)]  # 简化列名获取
                    diff.append(f"行不一致 PK={pk}, 列 {col}: MySQL值={m_val}, PG值={p_val}")
    return [f"表 {table} 数据不一致"] + diff if diff else []


def compare_data(mysql_conn, pg_conn, batch_size=1000):
    diff_report = []
    mysql_cur = mysql_conn.cursor(dictionary=True)
    pg_cur = pg_conn.cursor(cursor_factory=psycopg2.extras.DictCursor)

    mysql_tables = get_mysql_tables(mysql_conn)
    pg_tables = get_pg_tables(pg_conn)

    for table in mysql_tables:
        if table not in pg_tables:
            diff_report.append(f"表 {table} 不存在于目标库")
            continue

        try:
            mysql_pk = get_mysql_primary_keys(mysql_conn, table)
            if not mysql_pk:
                diff_report.append(f"表 {table} 无主键，跳过数据对比")
                continue
        except Exception as e:
            diff_report.append(f"表 {table} 主键获取失败: {str(e)}")
            continue

        offset = 0
        while True:
            try:
                mysql_data = fetch_mysql_data(mysql_cur, table, mysql_pk, batch_size, offset)
                print("mysql_data", mysql_data)
                pg_data = fetch_pg_data(pg_cur, table, mysql_pk, batch_size, offset)
                print("pg_data", pg_data)

                if pg_data is None:
                    diff_report.append(f"表 {table} 数据查询失败")
                    break

                batch_diff = find_data_diff(table, mysql_data, pg_data, mysql_pk)
                if batch_diff:
                    diff_report.extend(batch_diff)

                if len(mysql_data) < batch_size and len(pg_data) < batch_size:
                    break
                offset += batch_size
            except Exception as e:
                diff_report.append(f"表 {table} 对比异常: {str(e)}")
                break

    return diff_report


# 报告生成函数
def write_report(schema_diff, data_diff, output_dir="diff_reports"):
    os.makedirs(output_dir, exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    filepath = os.path.join(output_dir, f"diff_report_{timestamp}.txt")

    report = [
        "=" * 60,
        f"数据库对比报告 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        "=" * 60,
        f"源数据库: MySQL {MYSQL_CONFIG['host']}:{MYSQL_CONFIG['database']}",
        f"目标数据库: PostgreSQL {PG_CONFIG['host']}:{PG_CONFIG['database']}",
        "-" * 60
    ]

    report.append("\n[结构差异]")
    report.extend(schema_diff if schema_diff else ["无结构差异"])

    report.append("\n[数据差异]")
    report.extend(data_diff if data_diff else ["无数据差异"])

    try:
        with open(filepath, 'w', encoding='utf-8') as f:
            f.write('\n'.join(report))
        return filepath
    except Exception as e:
        print(f"报告生成失败: {str(e)}")
        return None


# 主程序
if __name__ == "__main__":
    start_time = datetime.now()

    mysql_conn = connect_mysql()
    pg_conn = connect_pg()

    print("开始结构对比...")
    schema_diff = compare_schemas(mysql_conn, pg_conn)

    print("\n开始数据对比...")
    data_diff = compare_data(mysql_conn, pg_conn)

    report_path = write_report(schema_diff, data_diff)

    duration = datetime.now() - start_time
    print("\n对比完成")
    print(f"总耗时: {duration.total_seconds():.2f}秒")
    if report_path:
        print(f"报告路径: {os.path.abspath(report_path)}")

    mysql_conn.close()
    pg_conn.close()


